package ex.tableapi.window;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.UNBOUNDED_ROW;

public class ExUnboundedOverWindows extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExUnboundedOverWindows demo = new ExUnboundedOverWindows();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");

        Table result = demo.exUnboundedOverWindows(orders);

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }

    private Table exUnboundedOverWindows(Table orders) {
        // Unbounded Event-time over window (assuming an event-time attribute "rowtime")
        Table result = orders.window(Over
                .partitionBy($("a"), $("c"))
                .orderBy($("rowtime"))
                .preceding(UNBOUNDED_ROW)
                .as("ow")) // 注意：此处不能使用 Expression 表达式
                .select($("a"),
                        $("c"),
                        $("rowtime"),
                        $("b").count().over($("ow")));
//proctime、rowtime；UNBOUNDED_ROW、UNBOUNDED_RANGE
        return result;
//        -- rowtime
//        UNBOUNDED_ROW
//        1> +I[liji, home, 2024-01-24T19:04:19, 1]
//        1> +I[liji, cloth, 2024-01-24T19:04:24, 1]
//        1> +I[liji, tv, 2024-01-24T19:35:51, 1]
//        1> +I[liji, home, 2024-01-25T19:04:19, 2]
//        1> +I[liji, home, 2024-01-25T19:04:19, 3]
//        1> +I[zy, tv, 2024-01-25T19:05:08, 1]
//        1> +I[zy, car, 2024-01-25T19:35:15, 1]
//        1> +I[zy, home, 2024-01-25T21:05:26, 1]
//        1> +I[zy, cloth, 2024-01-25T21:35:36, 1]
//        1> +I[zy, tv, 2024-01-26T19:05:07, 2]
//        --rowtime
//                ----UNBOUNDED_RANGE
//        1> +I[liji, home, 2024-01-24T19:04:19, 1]
//        1> +I[liji, cloth, 2024-01-24T19:04:24, 1]
//        1> +I[liji, tv, 2024-01-24T19:35:51, 1]
//        1> +I[liji, home, 2024-01-25T19:04:19, 3]
//        1> +I[liji, home, 2024-01-25T19:04:19, 3]
//        1> +I[zy, tv, 2024-01-25T19:05:08, 1]
//        1> +I[zy, car, 2024-01-25T19:35:15, 1]
//        1> +I[zy, home, 2024-01-25T21:05:26, 1]
//        1> +I[zy, cloth, 2024-01-25T21:35:36, 1]
//        1> +I[zy, tv, 2024-01-26T19:05:07, 2]
//
//
//        -- proctime
//        UNBOUNDED_RANGE：与当前行具有相同排序键的所有行都包含在窗口中
//        1> +I[liji, home, 2024-01-26T18:46:33, 3]
//        1> +I[liji, home, 2024-01-26T18:46:33, 3]
//        1> +I[liji, home, 2024-01-26T18:46:33, 3]
//        1> +I[zy, cloth, 2024-01-26T18:46:33, 1]
//        1> +I[zy, home, 2024-01-26T18:46:33, 1]
//        1> +I[zy, car, 2024-01-26T18:46:33, 1]
//        1> +I[zy, tv, 2024-01-26T18:46:33, 3]
//        1> +I[zy, tv, 2024-01-26T18:46:33, 3]
//        1> +I[zy, tv, 2024-01-26T18:46:33, 3]
//        1> +I[liji, tv, 2024-01-26T18:46:33, 1]
//        1> +I[liji, cloth, 2024-01-26T18:46:33, 1]
//        UNBOUNDED_ROW：上限设置为当前行
//        1> +I[liji, home, 2024-01-26T18:46:33, 1]
//        1> +I[liji, home, 2024-01-26T18:46:33, 2]
//        1> +I[liji, home, 2024-01-26T18:46:33, 3]
//        1> +I[zy, cloth, 2024-01-26T18:46:33, 1]
//        1> +I[zy, home, 2024-01-26T18:46:33, 1]
//        1> +I[zy, car, 2024-01-26T18:46:33, 1]
//        1> +I[zy, tv, 2024-01-26T18:46:33, 1]
//        1> +I[zy, tv, 2024-01-26T18:46:33, 2]
//        1> +I[zy, tv, 2024-01-26T18:46:33, 3]
//        1> +I[liji, tv, 2024-01-26T18:46:33, 1]
//        1> +I[liji, cloth, 2024-01-26T18:46:33, 1]
    }
}
